package demo.java.util.concurrent.lmax;

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.util.concurrent.Executors;

/**
 * LMAX Disruptor 是一种高性能的并发框架，最初由 LMAX 交易所开发，旨在处理高频交易中的数据流。Disruptor 通过独特的设计，可以在极低的延迟和高吞吐量下处理数据，从而提升程序的并发性能。
 * <p>
 * 主要特性
 * <li>高性能：Disruptor 旨在提供比传统的队列实现（如 BlockingQueue）更高的性能，通常能实现微秒级的延迟。
 * <li>低延迟：通过减少上下文切换和缓存失效的影响，Disruptor 可以显著降低线程间的通信延迟。
 * <li>单生产者或多生产者：Disruptor 支持单生产者和多生产者模式，可以灵活扩展以适应不同的场景。
 * <li>无锁设计：Disruptor 通过使用多个缓冲区和序列号来协调线程之间的访问，避免了锁的使用，减少了线程竞争带来的性能损失。
 * <p>
 * 主要组件
 * <li>Ring Buffer：这是 Disruptor 的核心数据结构。它是一个循环缓冲区，用于存储数据项。生产者将数据写入 Ring Buffer，消费者从中读取。
 * <li>序列号：每个数据项都有一个序列号，用于表示该数据项的顺序。消费者使用序列号来跟踪哪些数据项已被处理。
 * <li>事件：Disruptor 中的数据项被称为事件，生产者将事件发布到 Ring Buffer，消费者则处理这些事件。
 * <li>处理器：消费者实现了对事件的处理逻辑，可以是单一的或多个。</li>
 * <p>
 * 使用场景
 * LMAX Disruptor 适用于需要高频处理和极低延迟的场景，常见于：高频交易系统、实时数据处理、游戏服务器、事件驱动架构
 */
public class DisruptorDemo {

    private static final Logger logger = LoggerFactory.getLogger(DisruptorDemo.class);

    @Test
    public void test() {
        int bufferSize = 1024;

        // 创建工厂
        LongEventFactory factory = new LongEventFactory();

        // 创建 Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, Executors.defaultThreadFactory());

        // 注册处理器
        disruptor.handleEventsWith(new LongEventHandler());

        // 启动 Disruptor
        disruptor.start();

        // 生产者发布事件
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        for (long i = 0; i < 20; i++) {
            long sequence = ringBuffer.next(); // Grab the next sequence
            try {
                LongEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor
                event.set(i); // Fill with data
            } finally {
                ringBuffer.publish(sequence); // Publish
            }
        }

        disruptor.shutdown();
    }

    public static class LongEvent {
        private long value;

        public void set(long value) {
            this.value = value;
        }

        public long get() {
            return value;
        }
    }

    public static class LongEventFactory implements EventFactory<LongEvent> {
        public LongEvent newInstance() {
            return new LongEvent();
        }
    }

    public static class LongEventHandler implements EventHandler<LongEvent> {
        public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
            System.out.println("Event: " + event.get());
            logger.info("Event: {}", event.get());
        }
    }
}
